package com.jxb.service;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;

@Service
@Slf4j
public class CanalTestData {

    /*存储从表名到Redis缓存的键*/
    private Map<String,String> tableMapKey = new HashMap<>();

    @Autowired
    private CanalConnector connector;

    @Value("${canal.canal_test.subscribe:server}")
    private String subscribe;

    @Value("${canal.canal_test.batchSize}")
    private int batchSize;

    @PostConstruct
    public void connect() {
        tableMapKey.put("user","user");
        connector.connect();
        if("server".equals(subscribe))
            connector.subscribe(null);
        else
            connector.subscribe(subscribe);
        connector.rollback();
    }

    @PreDestroy
    public void disConnect() {
        connector.disconnect();
    }

    @Async
    @Scheduled(initialDelayString="${canal.promotion.initialDelay:5000}",fixedDelayString = "${canal.promotion.fixedDelay:5000}")
    public void processData() {
        try {
            if(!connector.checkValid()){
                log.warn("与Canal服务器的连接失效！！！重连，下个周期再检查数据变更");
                this.connect();
            }else{
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    log.info("本次【{}】没有检测到数据更新",batchId);
                }else {
                    log.info("本地【{}】有【{}】数据需要处理",batchId,size);
                    //一个表在一个周期内可能被修改多次，对缓存的处理只需要删除一次,Set集合可去重
                    Set<String> factKeys = new HashSet<>();
                    for (CanalEntry.Entry entry : message.getEntries()) {
                        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                        || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                            continue;
                        }
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                        String tableName = entry.getHeader().getTableName();
                        if (log.isDebugEnabled()) {
                            CanalEntry.EventType eventType = rowChange.getEventType();
                            log.debug("数据变更详情：来自binglog[{}.{}]，数据源{}.{}，变更类型{}",
                                    entry.getHeader().getLogfileName(),entry.getHeader().getLogfileOffset(),
                                    entry.getHeader().getSchemaName(),tableName,eventType);
                        }
                        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                            //获取前置镜像
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            log.info("--------------------------------------------------------前置镜像-----------------------------------------------------------------------");
                            for (CanalEntry.Column column : beforeColumnsList) {
                                log.info("columnName:{},columnValue:{}",column.getName(),column.getValue());
                            }
                            //获取后置镜像
                            log.info("--------------------------------------------------------后置镜像-----------------------------------------------------------------------");
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            for (CanalEntry.Column column : afterColumnsList) {
                                log.info("columnName:{},columnValue:{}",column.getName(),column.getValue());
                            }
                        }
                        factKeys.add(tableMapKey.get(tableName));
                    }
                    for (String key : factKeys) {
                        if (StringUtils.isNotEmpty(key)) {
                            log.info("【{}】触发修改缓存操作-----------------------redis------------------------------",key);
                        }
                    }
                    //提交确认
                    connector.ack(batchId);
                    log.info("本次【{}】处理数据完成",batchId);
                }
            }
        }catch (Exception e) {
            log.info("处理数据失败");
        }
    }

}
